package reactor;

import cn.hutool.core.util.RandomUtil;
import lombok.SneakyThrows;
import lombok.TestDto;
import org.junit.Test;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SynchronousSink;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.IntStream;
import java.util.stream.Stream;

/**
 * 异步
 *
 * @author qingyan 2022-01-19
 */
public class ReactorAysnc {
    /**
     * 设置睡眠时间
     */
    @SneakyThrows
    public static void randomSleep() {
        int randomInt = RandomUtil.randomInt(5);
        TimeUnit.SECONDS.sleep(randomInt);
        System.out.println("睡了" + randomInt + "秒");
    }

    public static void log(Object val) {
        System.out.println(Thread.currentThread().getName() + ":" + val.toString());
    }

    @SneakyThrows
    static void join() {
        Thread.currentThread().join();
    }

    @SneakyThrows
    @Test
    public void test() {
        Function<Integer, Integer> func = e -> {
            randomSleep();
            log(e);
            return e * e;
        };
        Mono<Integer> reduce = Flux.range(1, 100).map(func)
                .doOnNext(System.out::println).log().reduce((l, r) -> l += r);
        reduce.subscribeOn(Schedulers.newParallel("test", 30)).subscribe(System.out::println);

        join();

    }

    private Stream<String> getMovie() {
        System.out.println("Got the movie streaming request");
        return Stream.of(
                "scene 1",
                "scene 2",
                "scene 3",
                "scene 4",
                "scene 5"
        );
    }

    @SneakyThrows
    @Test
    public void test2() {
        //our NetFlux streamer
        //each scene will play for 2 seconds
        Flux<String> netFlux = Flux.fromStream(() -> getMovie())
                .delayElements(Duration.ofSeconds(1));

        // you start watching the movie
        netFlux.subscribe(scene -> System.out.println("You are watching " + scene));

        //I join after sometime
        TimeUnit.SECONDS.sleep(5);
        netFlux.subscribe(scene -> System.out.println("Vinsguru is watching " + scene));
        TimeUnit.SECONDS.sleep(5);
    }

    @SneakyThrows
    @Test
    public void test3() {
        //our movie theatre
        //each scene will play for 2 seconds
        Flux<String> movieTheatre = Flux.fromStream(() -> getMovie())
                .delayElements(Duration.ofSeconds(2)).share();

        // you start watching the movie
        movieTheatre.subscribe(scene -> System.out.println("You are watching " + scene));

        //I join after sometime
        Thread.sleep(5000);
        movieTheatre.subscribe(scene -> System.out.println("Vinsguru is watching " + scene));
        Thread.sleep(5000);
    }

    @SneakyThrows
    @Test
    public void test4() {
        //our movie theatre
        //each scene will play for 2 seconds
        Flux<String> movieTheatre = Flux.fromStream(() -> getMovie())
                .delayElements(Duration.ofSeconds(1)).share();

        // you start watching the movie
        movieTheatre.subscribe(scene -> System.out.println("You are watching " + scene));

        //I join after the source is completed
        Thread.sleep(6000);
        movieTheatre.subscribe(scene -> System.out.println("Vinsguru is watching " + scene));
        Thread.sleep(6000);
    }

    @SneakyThrows
    @Test
    public void test5() {
        //our movie theatre
        //each scene will play for 2 seconds
        Flux<String> movieTheatre = Flux.fromStream(() -> getMovie())
                .delayElements(Duration.ofSeconds(1)).share();

        // you start watching the movie
        movieTheatre.subscribe(scene -> System.out.println("You are watching " + scene));

        //I join after the source is completed
        Thread.sleep(6000);
        movieTheatre.subscribe(scene -> System.out.println("Vinsguru is watching " + scene));
        Thread.sleep(12000);
    }

    @SneakyThrows
    @Test
    public void test6() {
        //our movie theatre
        //each scene will play for 2 seconds
        Flux<Long> interval = Flux.interval(Duration.ofSeconds(1)).share();

        // you start watching the movie
        interval.subscribe(scene -> System.out.println("You are watching " + scene));

        //I join after the source is completed
        Thread.sleep(6000);
        interval.subscribe(scene -> System.out.println("Vinsguru is watching " + scene));
        Thread.sleep(12000);
    }

    @SneakyThrows
    @Test
    public void test7() {
        Flux<Integer> integerFlux = Flux.create((FluxSink<Integer> fluxSink) -> {
            IntStream.range(0, 5)
                    .peek(i -> System.out.println("going to emit - " + i))
                    .forEach(fluxSink::next);
        });

        //First observer. takes 1 ms to process each element
        integerFlux.delayElements(Duration.ofSeconds(1)).subscribe(i -> System.out.println("First :: " + i));

        //Second observer. takes 2 ms to process each element
        integerFlux.delayElements(Duration.ofSeconds(2)).subscribe(i -> System.out.println("Second:: " + i));
        Thread.sleep(12000);
    }

    @SneakyThrows
    @Test
    public void test8() {
        //create an instance of FluxSink implementation
        FluxSinkImpl fluxSinkConsumer = new FluxSinkImpl();

        //create method can accept this instance
        Flux<Integer> integerFlux = Flux.create(fluxSinkConsumer).delayElements(Duration.ofMillis(1)).share();
        integerFlux.delayElements(Duration.ofMillis(1)).subscribe(i -> System.out.println("First :: " + i));
        integerFlux.delayElements(Duration.ofMillis(2)).subscribe(i -> System.out.println("Second:: " + i));

        //We emit elements here
        IntStream.range(0, 5)
                .forEach(fluxSinkConsumer::publishEvent);

        TimeUnit.SECONDS.sleep(2);

    }

    @SneakyThrows
    @Test
    public void test9() {
        AtomicInteger atomicInteger = new AtomicInteger();

        //Flux generate sequence
        Flux<Integer> integerFlux = Flux.generate((SynchronousSink<Integer> synchronousSink) -> {
            System.out.println("Flux generate");
            synchronousSink.next(atomicInteger.getAndIncrement());
        });

        //observer
        integerFlux.delayElements(Duration.ofMillis(50))
                .subscribe(i -> System.out.println("First consumed ::" + i));

        TimeUnit.SECONDS.sleep(3);
    }

    @SneakyThrows
    @Test
    public void test10() {
        //To supply an initial state
        Callable<Integer> initialState = () -> 65;

        //BiFunction to consume the state, emit value, change state
        BiFunction<Integer, SynchronousSink<Character>, Integer> generator = (state, sink) -> {
            char value = (char)state.intValue();
            sink.next(value);
            if (value == 'Z') {
                sink.complete();
            }
            return state + 1;
        };

        //Flux which accepts initialstate and bifunction as arg
        Flux<Character> charFlux = Flux.generate(initialState, generator);

        //Observer
        charFlux.delayElements(Duration.ofMillis(50))
                .subscribe(i -> System.out.println("Consumed ::" + i));

        TimeUnit.SECONDS.sleep(5);
    }

    @Test
    public void test11(){
        Flux.range(1, 100)
                .log()
                .limitRate(10,3)
                .delayElements(Duration.ofMillis(100))
                .subscribe(System.out::println);
        join();
    }


    class FluxSinkImpl implements Consumer<FluxSink<Integer>> {

        private FluxSink<Integer> fluxSink;

        @Override
        public void accept(FluxSink<Integer> integerFluxSink) {
            this.fluxSink = integerFluxSink;
        }

        public void publishEvent(int event) {
            this.fluxSink.next(event);
        }

    }

    @Test
    public void test12(){
        Flux.range(1, 100)
                .log()
                .onBackpressureBuffer()
                .limitRate(10, 8)
                .subscribe();
        TestDto of = TestDto.of(1);
    }

}
